package rabbitmq.rpc;

import com.rabbitmq.client.*;
import rabbitmq.constant.Constants;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

/**
 * rpc
 * 1.发送请求
 * 2.接收响应
 */
public class RpcClient {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        //建立连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        //设置ip
        connectionFactory.setHost(Constants.HOST);
        //设置端口  需要提前开放端口号
        connectionFactory.setPort(Constants.PORT);
        //设置账号
        connectionFactory.setUsername(Constants.USERNAME);
        //设置密码
        connectionFactory.setPassword(Constants.PASSWORD);
        //设置虚拟机 要操作哪个虚拟机
        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);
        //创建连接
        Connection connection = connectionFactory.newConnection();
        //2.创建信道
        Channel channel = connection.createChannel();
        //声明交换机 这里使用内置的交换机

        //声明队列
        channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null);
        channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null);

        //发送消息
        //设置请求的一些属性
        String correlationId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties().builder()
                .correlationId(correlationId)
                .replyTo(Constants.RPC_RESPONSE_QUEUE)
                .build();
        String msg = "hello rabbitmq rpc";
        channel.basicPublish("", Constants.RPC_REQUEST_QUEUE, props, msg.getBytes());

        //接受响应，把响应结果放入到一个阻塞队列里面，从阻塞队列里面获取数据
        final BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(1);
        //响应逻辑
        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                //打印日志
                String responseMessage = new String(body);
                System.out.println("接收到回应消息:"+responseMessage);
                //进行判断
                if(correlationId.equals(properties.getCorrelationId())){
                    //将消息存储到阻塞队列里面
                    blockingQueue.offer(responseMessage);
                }
            }
        };

        channel.basicConsume(Constants.RPC_RESPONSE_QUEUE,true,consumer);
        //从阻塞队列里面进行获取
        String responseResult = blockingQueue.take();
        System.out.println("[RPC Clinent 响应结果]：" + responseResult);

    }
}
